package com.example.tms.realtime.app.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.tms.realtime.beans.TmsConfigDimBean;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MyBroadcastProcessFunction
 * Package: com.example.tms.realtime.app.fun
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-01
 * Time: 22:17
 */

public class MyBroadcastProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    private MapStateDescriptor<String, TmsConfigDimBean> mapStateDescriptor;

    public MyBroadcastProcessFunction(MapStateDescriptor<String, TmsConfigDimBean> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //处理主流业务的数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //TODO 主要就是用 广播流的数据和 主流的数据比对 如果是维度数据 那么就向下游传递 如果不是就过滤

        //获取操作业务数据库的表明
        String table = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TmsConfigDimBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据操作的业务数据哭的表明 到广播状态获取对应的配置信息
        TmsConfigDimBean tmsConfigDimBean = broadcastState.get(table);
        if (tmsConfigDimBean != null){
            //获取after对象 对应的是影响业务数据库表中的一条记录
            JSONObject after = jsonObj.getJSONObject("after");
            //过滤操作 过滤不需要的维度属性
            //过去过滤的依据
            String sinkColumns = tmsConfigDimBean.getSinkColumns();
            filterColumn(after,sinkColumns);

            //如果对应的配置信息不为空 是维度数据 向下游传递
            out.collect(after);
        }


    }

    private void filterColumn(JSONObject after, String sinkColumns) {
        String[] fieldArr = sinkColumns.split(",");
        List<String> filedList = Arrays.asList(fieldArr);
        Set<Map.Entry<String, Object>> entries = after.entrySet();
        entries.removeIf(entry->!filedList.contains(entry.getKey()));
    }

    //处理广播流的数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //{"before":null,"after":{"source_table":"line_base_shift","sink_table":"dim_line_base_shift","sink_family":null,"sink_columns":"id,line_id,start_time,est_end_time,driver1_emp_id,driver2_emp_id,truck_id,pair_shift_id,is_enabled,create_time,update_time,is_deleted","sink_pk":"id"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"tms_config","sequence":null,"table":"tms_config_dim","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1690899742446,"transaction":null}

        //将JSON字符串转换为JSON对象
        JSONObject jsonObject = JSON.parseObject(jsonStr);
        //获取广播状态 底层是单例的
        BroadcastState<String, TmsConfigDimBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);


        //获取配置表的操作类型
        String op = jsonObject.getString("op");
        if ("d".equals(op)){
            //如果从配置表中删除一条数据 从广播状态中 将对应的配置信息删除调用

            //获取表明 先获取到对象 在从对象中获取表明
            String sourceTable = jsonObject.getJSONObject("before").getString("source_table");
            broadcastState.remove(sourceTable);
        }else {
            //如果是删除之外的其他操作 将对应的配置信息放到或者更新到广播状态 将after对象 封装到 TmsBean对象
            TmsConfigDimBean tmsConfigDimBean = jsonObject.getObject("after", TmsConfigDimBean.class);
            //再从这个对象 获取维度表的表明
            String sinkTable = tmsConfigDimBean.getSinkTable();
            broadcastState.put(sinkTable,tmsConfigDimBean);

        }
    }
}